package org.jetlinks.pro.assets;

import lombok.Generated;
import lombok.NonNull;
import org.hswebframework.ezorm.core.Conditional;
import org.hswebframework.ezorm.core.NestConditional;
import org.hswebframework.ezorm.core.StaticMethodReferenceColumn;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.authorization.Authentication;
import org.hswebframework.web.authorization.DimensionType;
import org.hswebframework.web.authorization.exception.AccessDenyException;
import org.hswebframework.web.authorization.simple.SimpleDimensionType;
import org.jetlinks.core.utils.FluxUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 资产持有人信息,可通过持有人信息进行资产绑定信息获取,判断权限,实现编程式数据权限控制.
 *
 * <p>
 * 名词解释:
 * <ul>
 *     <li>资产: 在平台中表示某一种数据,例如: 一个设备,一个产品</li>
 *     <li>维度: 例如: 租户,部门,</li>
 * </ul>
 * <p>
 * 如果存在多个AssetsHolder,可以实现{@link org.springframework.core.Ordered}接口来设置优先级.
 * <p>
 * 如果指定来优先级,则在进行{@link AssetsHolder#hasPermission(String, Collection, boolean, AssetPermission...)}以及
 * {@link AssetsHolder#injectConditional(Conditional, String, String)}时,仅使用优先级最高的来处理.优先级相同则同时处理.
 *
 * <br>
 * <br>
 * <a href="http://doc.jetlinks.cn/dev-guide/assets.html">
 * 在线文档
 * </a>
 *
 * @author zhouhao
 * @see org.jetlinks.pro.assets.annotation.AssetsController
 * @see AssetsHolder#currentReactive()
 * @see AssetsHolder#assertPermission(String, Collection, boolean, AssetPermission...)
 * @see AssetsHolder#filterAssets(Flux, AssetType, Function, AssetPermission...)
 * @since 1.11
 */
@Generated
public interface AssetsHolder {

    /**
     * 默认排序
     */
    int ORDER_DEFAULT = 0;

    /**
     * 获取维度绑定信息,表示当前资产持有人持有哪些维度
     *
     * @return DimensionHolder
     */
    Flux<DimensionBinding> getDimensionBindings();

    /**
     * 获取指定维度的绑定信息
     *
     * @param targetType 类型
     * @return 绑定信息
     */
    Flux<DimensionBinding> getDimensionBindings(DimensionType targetType);

    /**
     * 获取指定ID的维度绑定信息
     *
     * @param targetType 维度类型
     * @param targetId   ID
     * @return 绑定信息
     */
    default Flux<DimensionBinding> getDimensionBindings(DimensionType targetType, Collection<?> targetId) {
        return this
            .getDimensionBindings(targetType)
            .filter(binding -> targetId.contains(binding.getTargetId()));
    }

    /**
     * 根据资产持有者重构topic以实现topic对数据权限控制
     *
     * @param topic topic
     * @return 重构后的topic列表
     */
    Flux<String> refactorTopic(String topic);

    /**
     * 获取当前持有者指定资产类型的全部资产绑定信息
     * <p>
     * 注意: 如果资产绑定到了多个维度,则会有多个绑定信息,如果仅对资产进行处理,请使用{@link Flux#distinct(Function)}进行去重
     *
     * @param assetType 资产类型
     * @return 资产集合
     */
    Flux<AssetBinding> getBindings(AssetType assetType);

    /**
     * 获取当前持有者指定资产类型的指定资产
     * <p>
     * 注意: 如果资产绑定到了多个维度,则会有多个绑定信息,如果仅对资产进行处理,请使用{@link Flux#distinct(Function)}进行去重
     *
     * @param assetType 资产类型
     * @param idList    资产ID
     * @return 资产集合
     */
    Flux<AssetBinding> getBindings(AssetType assetType, Collection<?> idList);

    /**
     * 获取指定维度和资产类型的全部资产绑定信息
     * <p>
     * 注意: 如果资产绑定到了多个维度,则会有多个绑定信息,如果仅对资产进行处理,请使用{@link Flux#distinct(Function)}进行去重
     *
     * @param targetType 维度类型
     * @param assetType  资产类型
     * @return 绑定信息
     */
    Flux<AssetBinding> getBindings(DimensionType targetType,
                                   AssetType assetType);

    /**
     * 获取指定维度和资产类型的资产绑定信息
     * <p>
     * 注意: 如果资产绑定到了多个维度,则会有多个绑定信息,如果仅对资产进行处理,请使用{@link Flux#distinct(Function)}进行去重
     *
     * @param targetType   维度类型
     * @param targetIdList 维度ID
     * @param assetType    资产类型
     * @return 绑定信息
     */
    Flux<AssetBinding> getBindings(DimensionType targetType,
                                   Collection<?> targetIdList,
                                   AssetType assetType);

    /**
     * 获取指定维度和资产类型的资产绑定信息
     * <p>
     * 注意: 如果资产绑定到了多个维度,则会有多个绑定信息,如果仅对资产进行处理,请使用{@link Flux#distinct(Function)}进行去重
     *
     * @param targetType   维度类型
     * @param targetIdList 维度ID
     * @param assetType    资产类型
     * @param assetIdList  资产ID
     * @return 绑定信息
     */
    Flux<AssetBinding> getBindings(DimensionType targetType,
                                   Collection<?> targetIdList,
                                   AssetType assetType,
                                   Collection<?> assetIdList);

    /**
     * 绑定资产到当前用户的维度中
     *
     * @param assetType 资产类型
     * @param assetId   资产ID
     * @return void
     */
    @Nonnull
    Mono<Void> bindAssets(@NonNull DimensionType targetType,
                          @NonNull AssetType assetType,
                          @NonNull Collection<?> assetId);

    /**
     * 绑定资产到指定的维度上
     *
     * @param assetType 资产类型
     * @param assetId   资产ID
     * @return void
     */
    @Nonnull
    default Mono<Void> bindAssets(@NonNull DimensionType targetType,
                                  @NonNull String targetId,
                                  @NonNull AssetType assetType,
                                  @NonNull Collection<?> assetId) {
        return bindAssets(targetType.getId(), targetId, assetType.getId(), assetId);
    }

    /**
     * 绑定资产到指定的维度
     *
     * <pre>
     *     //绑定设备到机构org1
     *     bindAssets("org","org1","device",deviceIdList)
     * </pre>
     *
     * @param assetType 资产类型
     * @param assetId   资产ID
     * @return void
     */
    @Nonnull
    Mono<Void> bindAssets(@NonNull String targetType,
                          @NonNull String targetId,
                          @NonNull String assetType,
                          @NonNull Collection<?> assetId);

    /**
     * 绑定指定资产类型的资产
     *
     * @param assetType 资产类型
     * @param assetId   资产Id
     * @return void
     */
    @Nonnull
    Mono<Void> bindAssets(@NonNull String assetType,
                          @NonNull Collection<?> assetId);


    /**
     * 绑定指定资产类型的资产
     *
     * @param assetType 资产类型
     * @param assetId   资产Id
     * @return void
     */
    @Nonnull
    default Mono<Void> bindAssets(@NonNull AssetType assetType,
                                  @NonNull Collection<?> assetId) {
        return bindAssets(assetType.getId(), assetId);
    }

    /**
     * 解除绑定指定资产类型的资产
     *
     * @param assetType 资产类型
     * @param assetId   资产Id
     * @return void
     */
    @Nonnull
    Mono<Void> unbindAssets(@NonNull String assetType,
                            @NonNull Collection<?> assetId);

    /**
     * 解除绑定资产到当前用户下
     *
     * @param assetType 资产类型
     * @param assetId   资产ID
     * @return void
     */
    @Nonnull
    default Mono<Void> unbindAssets(@NonNull String targetType,
                                    @NonNull String targetId,
                                    @NonNull AssetType assetType,
                                    @NonNull Collection<?> assetId) {
        return unbindAssets(targetType, targetId, assetType.getId(), assetId);
    }

    @Nonnull
    default Mono<Void> unbindAssets(@NonNull AssetType assetType,
                                    @NonNull Collection<?> assetId) {
        return unbindAssets(assetType.getId(), assetId);
    }

    /**
     * 解除绑定资产到当前用户下
     *
     * @param assetType 资产类型
     * @param assetId   资产ID
     * @return void
     */
    @Nonnull
    Mono<Void> unbindAssets(@NonNull String targetType,
                            @NonNull String targetId,
                            @NonNull String assetType,
                            @NonNull Collection<?> assetId);

    /**
     * 判定是否有指定资产的操作权限
     *
     * @param assetType  资产类型
     * @param assetId    资产ID集合
     * @param permission 权限
     * @return 是否有权限
     */
    @NonNull
    default Mono<Boolean> hasPermission(@NonNull AssetType assetType,
                                        @NonNull Collection<?> assetId,
                                        @NonNull AssetPermission... permission) {
        return hasPermission(assetType, assetId, false, permission);
    }

    /**
     * 判定是否有指定资产的操作权限
     *
     * @param assetType          资产类型
     * @param assetId            资产ID集合
     * @param permission         权限
     * @param allowAssetNotExist 如果资产不存在,是否返回true
     * @return 是否有权限
     */
    @NonNull
    default Mono<Boolean> hasPermission(@NonNull AssetType assetType,
                                        @NonNull Collection<?> assetId,
                                        boolean allowAssetNotExist,
                                        @NonNull AssetPermission... permission) {
        return hasPermission(assetType.getId(), assetId, allowAssetNotExist, permission);
    }

    /**
     * 判定是否有指定资产的操作权限,如果返回empty,Holder不处理此资产的权限判断
     *
     * @param assetType          资产类型
     * @param assetId            资产ID集合
     * @param permission         权限
     * @param allowAssetNotExist 如果资产不存在,是否返回true
     * @return 是否有权限
     */
    @NonNull
    Mono<Boolean> hasPermission(@NonNull String assetType,
                                @NonNull Collection<?> assetId,
                                boolean allowAssetNotExist,
                                @NonNull AssetPermission... permission);

    /**
     * 过滤掉没有权限的资产
     *
     * @param source        数据源
     * @param assetType     资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限
     * @param <T>           资产类型
     * @return 资产列表
     */
    @NonNull <T> Flux<T> filter(@NonNull Flux<T> source,
                                @NonNull AssetType assetType,
                                @NonNull Function<T, ?> assetIdMapper,
                                @NonNull AssetPermission... permission);

    /**
     * 判定是否有指定资产的操作权限
     *
     * @param assetType  资产类型 {@link AssetType#getId()}
     * @param assetId    资产ID集合
     * @param permission 权限
     * @return 是否有权限
     */
    @NonNull
    default Mono<Boolean> hasPermission(@NonNull String assetType,
                                        @NonNull Collection<?> assetId,
                                        @NonNull AssetPermission... permission) {
        return hasPermission(assetType, assetId, false, permission);
    }


    /**
     * 断言是否有指定资产的操作权限,如果没有权限则抛出{@link AccessDenyException}
     *
     * @param assetType  资产类型
     * @param assetId    资产ID
     * @param permission 权权限
     * @return void
     * @see AccessDenyException
     */
    default @NonNull Mono<Void> assetPermission(@NonNull AssetType assetType,
                                                @NonNull Collection<?> assetId,
                                                @NonNull AssetPermission... permission) {
        return assertPermission(assetType, assetId, false, permission);
    }

    /**
     * 断言是否有指定资产的操作权限,如果没有权限则抛出{@link AccessDenyException}
     *
     * @param assetType          资产类型
     * @param assetId            资产ID
     * @param permission         权权限
     * @param allowAssetNotExist 如果资产不存在,是否返回true
     * @return void
     * @see AccessDenyException
     */
    default @NonNull Mono<Void> assertPermission(@NonNull AssetType assetType,
                                                 @NonNull Collection<?> assetId,
                                                 boolean allowAssetNotExist,
                                                 @NonNull AssetPermission... permission) {
        return hasPermission(assetType, assetId, allowAssetNotExist, permission)
            .filter(Boolean.TRUE::equals)
            .switchIfEmpty(Mono.error(AccessDenyException::new))
            .then();
    }

    /**
     * 断言是否有指定资产的操作权限,如果没有权限则抛出{@link AccessDenyException}
     *
     * @param assetType          资产类型
     * @param assetId            资产ID
     * @param permission         权权限
     * @param allowAssetNotExist 如果资产不存在,是否返回true
     * @return void
     * @see AccessDenyException
     */
    default @NonNull Mono<Void> assertPermission(@NonNull String assetType,
                                                 @NonNull Collection<?> assetId,
                                                 boolean allowAssetNotExist,
                                                 @NonNull AssetPermission... permission) {
        return hasPermission(assetType, assetId, allowAssetNotExist, permission)
            .filter(Boolean.TRUE::equals)
            .switchIfEmpty(Mono.error(AccessDenyException::new))
            .then();
    }

    /**
     * 断言是否有指定资产的操作权限,如果没有权限则抛出{@link AccessDenyException}
     *
     * @param assetType  资产类型
     * @param assetId    资产ID
     * @param permission 权权限
     * @return void
     * @see AccessDenyException
     */
    default @NonNull Mono<Void> assertPermission(@NonNull String assetType,
                                                 @NonNull Collection<?> assetId,
                                                 @NonNull AssetPermission... permission) {
        return assertPermission(assetType, assetId, false, permission);
    }


    /**
     * 注入数据权限控制动态条件(嵌套)
     *
     * @param conditional 动态条件
     * @param assetType   资产类型 {@link AssetType#getId()}
     * @param idProperty  资产主键属性 如: id
     * @param <T>         动态条件类型
     * @return 注入后的动态条件
     */
    <T extends Conditional<T>> NestConditional<T> injectConditional(NestConditional<T> conditional,
                                                                    String assetType,
                                                                    String idProperty);

    /**
     * 注入数据权限控制动态条件
     *
     * @param conditional 动态条件
     * @param assetType   资产类型 {@link AssetType#getId()}
     * @param idProperty  资产主键属性 如: id
     * @param <T>         动态条件类型
     * @return 注入后的动态条件
     */
    default <T extends Conditional<T>> T injectConditional(T conditional,
                                                           String assetType,
                                                           String idProperty) {
        return conditional.nest(nest -> this.injectConditional(nest, assetType, idProperty));
    }

    /**
     * 将租户数据权限控制条件注入到动态查询参数
     *
     * @param query      查询参数
     * @param assetType  资产类型 {@link AssetType#getId()}
     * @param idProperty 资产主键属性 如: id
     * @return 注入后的查询参数
     */
    default QueryParamEntity injectQuery(QueryParamEntity query,
                                         String assetType,
                                         String idProperty) {
        return injectConditional(query.toNestQuery(), assetType, idProperty).getParam();
    }

    /**
     * 将租户数据权限控制条件注入到动态查询参数
     *
     * @param query      查询参数
     * @param assetType  资产类型 {@link AssetType#getId()}
     * @param idProperty 资产主键属性 使用方法引用来定义属性,如: Entity::getId
     * @return 注入后的查询参数
     */
    default <T> QueryParamEntity injectQuery(QueryParamEntity query,
                                             String assetType,
                                             StaticMethodReferenceColumn<T> idProperty) {
        return injectQuery(query, assetType, idProperty.getColumn());
    }

    default <T> QueryParamEntity injectQuery(QueryParamEntity query,
                                             AssetType assetType,
                                             StaticMethodReferenceColumn<T> idProperty) {
        return injectQuery(query, assetType.getId(), idProperty);
    }

    /**
     * 将租户数据权限控制条件注入到异步动态查询参数
     *
     * @param query      查询参数
     * @param assetType  资产类型 {@link AssetType#getId()}
     * @param idProperty 资产主键属性 使用方法引用来定义属性,如: Entity::getId
     * @return 注入后的查询参数
     */
    default Mono<QueryParamEntity> injectQuery(Mono<QueryParamEntity> query,
                                               String assetType,
                                               String idProperty) {
        return query.map(q -> injectQuery(q, assetType, idProperty));
    }

    /**
     * 将租户数据权限控制条件注入到异步动态查询参数
     *
     * @param query      查询参数
     * @param assetType  资产类型 {@link AssetType#getId()}
     * @param idProperty 资产主键属性 使用方法引用来定义属性,如: Entity::getId
     * @return 注入后的查询参数
     */
    default <T> Mono<QueryParamEntity> injectQuery(Mono<QueryParamEntity> query,
                                                   String assetType,
                                                   StaticMethodReferenceColumn<T> idProperty) {
        return injectQuery(query, assetType, idProperty.getColumn());
    }


    /**
     * 断言用户拥有对资产的权限,如果用户不是资产持有者成员则直接通过,如果没有任意一个资产信息,则抛出异常{@link AccessDenyException}
     *
     * @param source             数据源
     * @param type               资产类型
     * @param assetIdMapper      资产ID转换
     * @param allowAssetNotExist 资产不存在时,默认通过
     * @param permission         权限列表
     * @param <T>                资产数据类型
     * @return 源数据
     * @see AccessDenyException
     */
    static <T> Flux<T> assertPermission(Flux<T> source,
                                        AssetType type,
                                        Function<T, ?> assetIdMapper,
                                        boolean allowAssetNotExist,
                                        AssetPermission... permission) {
        Flux<T> cache = source.cache();
        return Mono
            .zip(AssetsHolder.currentReactive(),
                 cache.as(FluxUtils.safeMap(assetIdMapper)).collect(Collectors.toSet()))
            .flatMapMany(tp2 -> tp2
                .getT1()
                .assertPermission(type, tp2.getT2(), allowAssetNotExist, permission))
            .thenMany(cache);
    }

    /**
     * 断言用户拥有对资产的权限,如果没有权限,则抛出异常{@link AccessDenyException}
     *
     * @param source             数据源
     * @param type               资产类型
     * @param assetIdMapper      资产ID转换
     * @param allowAssetNotExist 资产不存在时,默认通过
     * @param permission         权限列表
     * @param <T>                资产数据类型
     * @return 源数据
     * @see AccessDenyException
     */
    static <T> Mono<T> assertPermission(Mono<T> source,
                                        AssetType type,
                                        Function<T, ?> assetIdMapper,
                                        boolean allowAssetNotExist,
                                        AssetPermission... permission) {
        Mono<T> cache = source.cache();
        return Mono
            .zip(AssetsHolder.currentReactive(), cache)
            .flatMap(tp2 -> tp2
                .getT1()
                .assertPermission(type, Collections.singleton(assetIdMapper.apply(tp2.getT2())), allowAssetNotExist, permission)
                .thenReturn(tp2.getT2()))
            .switchIfEmpty(cache);
    }


    /**
     * 断言当前资产持有者对指定数据有相应的权限
     *
     * @param source        数据源
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限,空数组则只不判断权限,只要有资产即通过
     * @param <T>           数据类型
     * @return 处理后的数据流
     * @see AccessDenyException
     */
    static <T> Mono<T> assertPermission(Mono<T> source,
                                        AssetType type,
                                        Function<T, ?> assetIdMapper,
                                        AssetPermission... permission) {
        return assertPermission(source, type.getId(), assetIdMapper, permission);
    }

    /**
     * 判断指定的资产持有者成员是否有资产权限
     *
     * @param memberMono    成员
     * @param source        资产数据流
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换
     * @param permission    权限
     * @param <T>           资产类型
     * @return 资产流
     * @since 1.9
     */
    static <T> Mono<T> assertPermission(Mono<AssetsHolder> memberMono,
                                        Mono<T> source,
                                        String type,
                                        Function<T, ?> assetIdMapper,
                                        AssetPermission... permission) {
        Mono<T> cache = source.cache();
        return Mono
            .zip(memberMono, cache)
            .flatMap(tp2 -> tp2
                .getT1()
                .assertPermission(type, Collections.singleton(assetIdMapper.apply(tp2.getT2())), permission)
                .thenReturn(tp2.getT2()))
            .switchIfEmpty(cache);
    }

    /**
     * 断言当前资产持有者对指定数据有相应的权限
     *
     * @param source        数据源
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限,空数组则只不判断权限,只要有资产即通过
     * @param <T>           数据类型
     * @return 处理后的数据流
     * @see AccessDenyException
     */
    static <T> Mono<T> assertPermission(Mono<T> source,
                                        String type,
                                        Function<T, ?> assetIdMapper,
                                        AssetPermission... permission) {
        return assertPermission(AssetsHolder.currentReactive(), source, type, assetIdMapper, permission);
    }

    /**
     * 断言当前资产持有者对指定数据Id的数据有相应的权限
     *
     * @param type       资产类型
     * @param id         数据ID
     * @param permission 权限,空数组则只不判断权限,只要有资产即通过
     * @return void
     * @see AccessDenyException
     */
    static Mono<Void> assertPermission(String type,
                                       Object id,
                                       AssetPermission... permission) {
        return assertPermission(Mono.just(id), type, Function.identity(), permission).then();
    }


    /**
     * 断言当前资产持有者对指定数据Id的数据有相应的权限
     *
     * @param type       资产类型
     * @param id         数据ID
     * @param permission 权限,空数组则只不判断权限,只要有资产即通过
     * @return void
     * @see AccessDenyException
     */
    static Mono<Void> assertPermission(AssetType type,
                                       Object id,
                                       AssetPermission... permission) {
        return assertPermission(Mono.just(id), type, Function.identity(), permission).then();
    }

    /**
     * 断言当前资产持有者对指定数据Id的数据有相应的权限
     *
     * @param type       资产类型
     * @param id         数据ID
     * @param permission 权限,空数组则只不判断权限,只要有资产即通过
     * @return void
     * @see AccessDenyException
     */
    static Mono<Void> assertPermission(AssetType type,
                                       Collection<Object> id,
                                       AssetPermission... permission) {
        return assertPermission(Flux.fromIterable(id), type, Function.identity(), permission).then();
    }


    /**
     * 断言当前资产持有者数据源中数据的相应权限,如果当前用户不是资产持有者,则默认通过.
     *
     * @param source        数据源
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限,空数组则只不判断权限,只要有资产即通过
     * @param <T>           数据类型
     * @return 处理后的数据流
     * @see AccessDenyException
     */
    static <T> Flux<T> assertPermission(Flux<T> source,
                                        AssetType type,
                                        Function<T, ?> assetIdMapper,
                                        AssetPermission... permission) {
        return assertPermission(AssetsHolder.currentReactive(), source, type, assetIdMapper, permission);
    }

    /**
     * 断言指定资产持有者拥有数据的相应权限.
     *
     * @param memberMono    资产持有者成员
     * @param source        数据源
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限,空数组则只不判断权限,只要有资产即通过
     * @param <T>           数据类型
     * @return 处理后的数据流
     * @see AccessDenyException
     */
    static <T> Flux<T> assertPermission(Mono<AssetsHolder> memberMono,
                                        Flux<T> source,
                                        AssetType type,
                                        Function<T, ?> assetIdMapper,
                                        AssetPermission... permission) {
        Flux<T> cache = source.cache();
        return Mono
            .zip(memberMono,
                 cache.as(FluxUtils.safeMap(assetIdMapper)).collect(Collectors.toSet()))
            .flatMapMany(tp2 -> tp2
                .getT1()
                .assetPermission(type, tp2.getT2(), permission))
            .thenMany(cache);
    }

    /**
     * 绑定资产当前用户的指定维度中
     *
     * @param source        数据流
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param <T>           数据类型
     * @return 原始数据
     */
    static <T> Flux<T> bindAssets(Flux<T> source,
                                  DimensionType dimensionType,
                                  AssetType type,
                                  Function<T, ?> assetIdMapper) {
        Flux<T> cache = source.cache();
        return Mono
            .zip(AssetsHolder.currentReactive(),
                 cache.as(FluxUtils.safeMap(assetIdMapper)).collect(Collectors.toSet()))
            .flatMapMany(tp2 -> tp2.getT1().bindAssets(dimensionType, type, tp2.getT2()))
            .thenMany(cache);
    }


    /**
     * 绑定资产到当前租户中
     *
     * @param source        数据流
     * @param type          资产类型
     * @param assetIdMapper 资产ID转换器
     * @param <T>           数据类型
     * @return 原始数据
     */
    static <T> Flux<T> bindAssets(Flux<T> source,
                                  AssetType type,
                                  Function<T, ?> assetIdMapper) {
        Flux<T> cache = source.cache();
        return Mono
            .zip(AssetsHolder.currentReactive(),
                 cache.as(FluxUtils.safeMap(assetIdMapper)).collect(Collectors.toSet()))
            .flatMap(tp2 -> tp2.getT1().bindAssets(type, tp2.getT2()))
            .thenMany(cache);
    }


    /**
     * 给动态查询条件中注入资产数据过滤查询条件.
     *
     * @param queryParamEntity 动态查询条件
     * @param type             资产类型
     * @param property         属性
     * @return 修改后对查询条件
     */
    static Mono<QueryParamEntity> injectQueryParam(QueryParamEntity queryParamEntity,
                                                   AssetType type,
                                                   String property) {
        return injectQueryParam(queryParamEntity, type.getId(), property);
    }

    /**
     * 给动态查询条件中注入资产数据过滤查询条件.
     *
     * @param queryParamEntity 动态查询条件
     * @param type             资产类型
     * @param property         静态方法引用属性
     * @return 修改后对查询条件
     */
    static <E> Mono<QueryParamEntity> injectQueryParam(QueryParamEntity queryParamEntity,
                                                       AssetType type,
                                                       StaticMethodReferenceColumn<E> property) {
        return injectQueryParam(queryParamEntity, type.getId(), property.getColumn());
    }

    static <E> Mono<QueryParamEntity> injectQueryParam(Mono<AssetsHolder> holderMono,
                                                       QueryParamEntity queryParamEntity,
                                                       AssetType type,
                                                       StaticMethodReferenceColumn<E> property) {
        return holderMono
            .map(holder -> holder.injectQuery(queryParamEntity, type.getId(), property))
            .defaultIfEmpty(queryParamEntity);
    }

    /**
     * 给动态查询条件中注入资产数据过滤查询条件.
     *
     * @param holderMono       资产持有人
     * @param queryParamEntity 动态查询条件
     * @param type             资产类型
     * @param property         属性
     * @return 修改后对查询条件
     */
    static Mono<QueryParamEntity> injectQueryParam(Mono<AssetsHolder> holderMono,
                                                   QueryParamEntity queryParamEntity,
                                                   String type,
                                                   String property) {
        return holderMono
            .map(holder -> holder.injectQuery(queryParamEntity, type, property))
            .defaultIfEmpty(queryParamEntity);
    }

    /**
     * 根据动态条件注入数据权限控制条件,并返回注入动态条件后的条件.
     * <pre>
     *     AssetsHolder
     *     .injectConditional(
     *          createQuery().where(MyEntity::getName,name),
     *          MyAssetType.type,
     *          MyEntity::getId)
     *     .flatMap(ReactiveQuery::fetch)
     * </pre>
     *
     * @param condition Conditional
     * @param type      资产类型
     * @param property  属性
     * @param <T>       条件类型
     * @return async Conditional
     */
    static <T extends Conditional<T>, R> Mono<T> injectConditional(T condition,
                                                                   AssetType type,
                                                                   StaticMethodReferenceColumn<R> property) {
        return injectConditional(AssetsHolder.currentReactive(), condition, type.getId(), property.getColumn());
    }

    /**
     * 根据动态条件注入资产持有者控制条件,并返回注入动态条件后的条件.
     * <pre>
     *     AssetsHolder
     *     .injectConditional(
     *          AssetsHolder.currentReactive(),
     *          createQuery().where(MyEntity::getName,name),
     *          "type",
     *          "id")
     *     .flatMap(ReactiveQuery::fetch)
     * </pre>
     *
     * @param condition Conditional
     * @param type      资产类型
     * @param property  属性
     * @param <T>       条件类型
     * @return async Conditional
     */
    static <T extends Conditional<T>> Mono<T> injectConditional(Mono<AssetsHolder> holderMono,
                                                                T condition,
                                                                String type,
                                                                String property) {
        return holderMono
            .map(holder -> holder.injectConditional(condition, type, property))
            .defaultIfEmpty(condition);
    }

    /**
     * 给动态查询条件中注入资产数据过滤查询条件.
     *
     * @param queryParamEntity 动态查询条件
     * @param type             资产类型
     * @param property         属性
     * @return 修改后对查询条件
     */
    static Mono<QueryParamEntity> injectQueryParam(QueryParamEntity queryParamEntity,
                                                   String type,
                                                   String property) {
        return injectQueryParam(AssetsHolder.currentReactive(), queryParamEntity, type, property);
    }

    /**
     * 过滤掉当前资产持有者没有权限的资产
     *
     * @param source        数据源
     * @param assetType     资产类型
     * @param assetIdMapper 资产ID转换器
     * @param permission    权限
     * @param <T>           资产类型
     * @return 资产列表
     */
    static <T> Flux<T> filterAssets(Flux<T> source,
                                    AssetType assetType,
                                    Function<T, ?> assetIdMapper,
                                    AssetPermission... permission) {
        return AssetsHolder
            .currentReactive()
            .map(tenant -> tenant.filter(source, assetType, assetIdMapper, permission))
            .defaultIfEmpty(source)
            .flatMapMany(Function.identity());
    }


    /**
     * 获取当前资产持有人信息,通常是当前登录用户的资产信息
     *
     * @return AssetsHolder
     * @see AssetsHolderProvider
     */
    static Mono<AssetsHolder> currentReactive() {
        return AssetsHolderHolder.current();
    }

    /**
     * 获取所有的资产持有人信息
     *
     * @return AssetsHolder
     */
    static Flux<AssetsHolder> currentReactiveAll() {
        return AssetsHolderHolder.currentAll();
    }

    /**
     * 根据权限信息获取资产持有人信息
     *
     * @return AssetsHolder
     */
    static Mono<AssetsHolder> fromAuth(Authentication authentication) {
        return AssetsHolderHolder.getHolder(authentication);
    }

    static Flux<AssetsHolder> fromAuthAll(Authentication authentication) {
        return AssetsHolderHolder.getHolders(authentication);
    }

    /**
     * 过滤当前资产持有人拥有维度的资产权限的资产
     *
     * @param source           要过滤的数据流源
     * @param targetTypeGetter 获取绑定维度类型函数
     * @param targetIdGetter   获取绑定维度ID函数
     * @param assetTypeGetter  获取资产类型函数
     * @param assetIdGetter    获取绑定资产ID函数
     * @param <T>              数据类型
     * @return 过滤后的资产类型
     */
    static <T> Flux<T> filter(Flux<T> source,
                              Function<T, DimensionType> targetTypeGetter,
                              Function<T, Collection<String>> targetIdGetter,
                              Function<T, AssetType> assetTypeGetter,
                              Function<T, Collection<String>> assetIdGetter,
                              AssetPermission... permissions) {
        return filter(currentReactive(),
                      source,
                      targetTypeGetter,
                      targetIdGetter,
                      assetTypeGetter,
                      assetIdGetter,
                      permissions);
    }

    /**
     * 过滤指定资产持有人拥有维度的资产权限的资产
     *
     * @param source           要过滤的数据流源
     * @param targetTypeGetter 获取绑定维度类型函数
     * @param targetIdGetter   获取绑定维度ID函数
     * @param assetTypeGetter  获取资产类型函数
     * @param assetIdGetter    获取绑定资产ID函数
     * @param <T>              数据类型
     * @return 过滤后的资产类型
     */
    static <T> Flux<T> filter(Mono<AssetsHolder> holderMono,
                              Flux<T> source,
                              Function<T, DimensionType> targetTypeGetter,
                              Function<T, Collection<String>> targetIdGetter,
                              Function<T, AssetType> assetTypeGetter,
                              Function<T, Collection<String>> assetIdGetter,
                              AssetPermission... permissions) {
        Flux<T> cache = source.cache();
        return holderMono
            .map(holder -> cache
                .filterWhen(request -> Mono
                    .zip(
                        //有维度
                        holder.getDimensionBindings(
                                  targetTypeGetter.apply(request),
                                  targetIdGetter.apply(request)
                              )
                              .hasElements(),
                        //有权限
                        holder.hasPermission(
                            assetTypeGetter.apply(request),
                            assetIdGetter.apply(request),
                            false,
                            permissions
                        ),
                        (hasDimension, hasPermission) -> hasDimension && hasPermission
                    )))
            .defaultIfEmpty(cache)
            .flatMapMany(Function.identity());
    }

    /**
     * 使用指定的资产持有人来重构topic,如果不存在资产持有人则返回原始topic
     *
     * @param holderMono 资产持有人
     * @param topic      topic
     * @return 重构后的topic
     */
    static Flux<String> refactorTopic(Mono<AssetsHolder> holderMono, String topic) {
        return holderMono
            .map(holder -> holder.refactorTopic(topic))
            .defaultIfEmpty(Flux.just(topic))
            .flatMapMany(Function.identity());
    }

}
